// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jhkim

#pragma once
#include <algorithm>
#include <functional>
#include <memory>
#include <queue>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "coin2/exchange/order/system.pb.h"
#include "coin2/exchange/sim/order_v1/light_book.h"
#include "coin2/exchange/sim/order_v1/matching_engine/matching_engine.h"
#include "coin2/exchange/sim/order_v1/matching_engine/queue_status.h"
#include "fastfeature/simulator_spec.pb.h"
#include "presto/quant/feature/util/coin_sweep2.h"
#include "presto/quant/feature/util/decimal.h"
#include "presto/quant/math/moving_average.h"
#include "presto/quant/symbology/symbol_map.h"

namespace coin2::exchange::sim::order_v1 {

using coin2::exchange::order::OrderRecordProto;
using coin2::exchange::order::AccountSimConfig;

typedef std::set<std::shared_ptr<QueueStatus>, QueueStatusComparator> SortedQs;

class ResponsePublisher {
 public:
  void PublishFill(
      int64_t order_id,
      int64_t time,
      int sign,
      double order_price,
      double order_qty,
      bool post_only,
      std::string tag,
      OrderRecordProto::FillType fill_type,
      double fill_price,
      double fill_qty) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(OrderRecordProto::ORDER_FILLED);
    record->set_order_id(order_id);
    record->set_time(time);
    record->set_sign(sign);
    record->set_order_price(order_price);
    record->set_order_qty(order_qty);
    if (write_tag) record->set_tag(tag);
    record->set_post_only(post_only);
    record->set_fill_type(fill_type);
    record->set_fill_price(fill_price);
    record->set_fill_qty(fill_qty);
    working_qty_sign[sign] -= fill_qty;
  }

  void PublishSubmitAcceptOrder(
      int64_t order_id,
      int64_t time,
      int64_t submit_time,
      double price,
      double qty,
      int sign,
      bool post_only,
      std::string tag,
      OrderRecordProto::RecordType record_type,
      std::optional<double> submit_max_rate_used_ratio,
      std::optional<double> cancel_max_rate_used_ratio) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(record_type);
    record->set_order_id(order_id);
    record->set_time(time);
    record->set_submit_time(submit_time);
    record->set_order_price(price);
    record->set_order_qty(qty);
    if (write_tag) record->set_tag(tag);
    record->set_sign(sign);
    record->set_post_only(post_only);
    if (submit_max_rate_used_ratio.has_value()) {
      record->set_submit_max_rate_used_ratio(*submit_max_rate_used_ratio);
    }
    if (cancel_max_rate_used_ratio.has_value()) {
      record->set_cancel_max_rate_used_ratio(*cancel_max_rate_used_ratio);
    }
    if (record_type == OrderRecordProto::ORDER_ACCEPTED) {
      working_qty_sign[sign] += qty;
    }
  }

  void PublishOrderError(
      int64_t order_id,
      int64_t time,
      int64_t submit_time,
      double price,
      double qty,
      int sign) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(OrderRecordProto::ORDER_ERROR);
    record->set_order_id(order_id);
    record->set_time(time);
    record->set_submit_time(submit_time);
    record->set_order_price(price);
    record->set_order_qty(qty);
    record->set_sign(sign);
    working_qty_sign[sign] -= qty;
  }

  void PublishCancelConfirm(
      int64_t order_id,
      int64_t time,
      int64_t cancel_submit_time,
      double order_price,
      double cancel_qty,
      int sign) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(OrderRecordProto::CANCEL_CONFIRMED);
    record->set_order_id(order_id);
    record->set_order_price(order_price);
    record->set_cancel_qty(cancel_qty);
    record->set_time(time);
    record->set_submit_time(cancel_submit_time);
    record->set_sign(sign);
    working_qty_sign[sign] -= cancel_qty;
  }

  void PublishCancelSubmitted(int64_t order_id, int64_t time, int64_t cancel_submit_time) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(OrderRecordProto::CANCEL_SUBMITTED);
    record->set_order_id(order_id);
    record->set_time(time);
  }

  void PublishCancelError(
      int64_t order_id,
      int64_t time,
      int64_t cancel_submit_time,
      double order_price,
      int sign) {
    records.resize(records.size() + 1);
    auto* record = &records.back();
    record->set_type(OrderRecordProto::CANCEL_ERROR);
    record->set_order_id(order_id);
    record->set_order_price(order_price);
    record->set_submit_time(cancel_submit_time);
    record->set_time(time);
    record->set_sign(sign);
  }

  const std::vector<OrderRecordProto>& GetOrderRecords() const { return records; }
  void ClearOrderRecords() { records.clear(); }

  mutable std::unordered_map<int, double> working_qty_sign;
  bool write_tag{false};

 private:
  std::vector<OrderRecordProto> records;
};

struct QueueTracker {
  void ApplyFeed(const ::fastfeature::FeedMessage& feed, std::shared_ptr<QueueStatus> qs) {
    if (qs->batch_trade) {
      if (!qs->batch_trade->WithinBatch(feed.fetched_time())) {
        qs->batch_trade->MatchAndFlushBatch(qs.get());
        qs->ApplyCancel(feed.fetched_time());
      }
    } else {
      CHECK_EQ((qs->cancel_arrival_time > 0), (qs->cancel_confirmed_time > 0));
    }
    if (feed.feed_type() == ::fastfeature::TRADE_FEED) {
      if (qs->batch_trade && qs->batch_trade->WithinBatch(feed.fetched_time())) {
        qs->batch_trade->AddBatch(feed);
        return;
      }
      qs->ApplyTrade(feed);
    } else if (feed.feed_type() == ::fastfeature::BOOK_FEED) {
      lb.UpdateBook(feed);
      qs->ApplyBook(lb);
    }
  }
  LightBook_10_dec12 lb;
};

class SymbolMatchingEngine {
 public:
  SymbolMatchingEngine() = default;
  explicit SymbolMatchingEngine(const AccountSimConfig& acct_config)
      : is_snapshot(acct_config.is_snapshot()),
        use_exchange_timestamp_(acct_config.use_exchange_timestamp()),
        makerfill_tolerance_bps(acct_config.makerfill_tolerance_bps()),
        bboq_window(600L * 1e9L),
        position(acct_config.initial_position()) {
    book = std::make_unique<LightBook_10_dec12>(use_exchange_timestamp_);
    qtrack = std::make_unique<QueueTracker>();
    publisher_.write_tag = acct_config.write_tag();
  }
  void SubmitCancel(int64_t order_id, int64_t cancel_submit_time);
  void SubmitAcceptOrder(
      int64_t order_id,
      int64_t submit_time,
      double price,
      double qty,
      int sign,
      bool post_only,
      std::string tag,
      OrderRecordProto::RecordType record_type,
      std::optional<double> submit_max_rate_used_ratio,
      std::optional<double> cancel_max_rate_used_ratio
      );
  const std::vector<OrderRecordProto>& GetOrderRecords() const;
  void ClearOrderRecords();
  void UpdateFeed(const Symbol* symbol, const ::fastfeature::FeedMessage& feed);
  void SendOrder(
      int64_t order_arrival_time,
      const PriceType& price,
      double order_qty,
      int sign,
      bool post_only,
      std::string tag,
      bool ioc,
      int64_t order_id);
  void CancelOrder(int64_t order_id, int64_t cancel_arrival_time);
  double GetPosition() const;
  void SetExchangeTime(int64_t time);
  void SetFetchedTime(int64_t ftime);
  int64_t GetFetchedTime() const;
  void AcceptQueueStatus(QueueStatus* qs, int64_t timestamp);
  void ClearDeadOrders();
  void CancelOrderActual(int64_t order_id, int64_t cancel_confirm_time);
  double GetWorkingQty(int sign) const {
    return publisher_.working_qty_sign[sign];
  }
  double GetMtmPrice() const { return mtm_price; }
  std::pair<double, double> GetWorkingBbo() const {
    std::pair<double, double> bbo;
    if (!buys.empty()) bbo.first = (*buys.cbegin())->price;
    if (!sells.empty()) bbo.second = (*sells.cbegin())->price;
    return bbo;
  }

  int64_t time{};
  int64_t ftime{};
 private:
  const bool is_snapshot{};
  const bool use_exchange_timestamp_{false};
  const double makerfill_tolerance_bps{};
  presto::math::TimeWindowMovingAverage<double> bboq_window;
  std::unique_ptr<LightBook_10_dec12> prebook;
  std::unique_ptr<LightBook_10_dec12> book;
  std::unique_ptr<QueueTracker> qtrack;
  SortedQs buys;
  SortedQs sells;
  std::vector<std::shared_ptr<QueueStatus>> dead_orders;
  std::unordered_map<int64_t, std::shared_ptr<QueueStatus>> map_on_oid;
  double position{};
  double mtm_price{};
  ResponsePublisher publisher_;
};


struct JobQueue {
  void PushElement(int64_t submit_time, int64_t current_time, const std::function<void()>& job) {
    jobs.emplace(std::make_pair(submit_time, job));
  }

  void UpdateTimeOnly(int64_t time) {
    while (!jobs.empty() && jobs.top().first < time) {
      jobs.top().second.operator()();
      jobs.pop();
    }
  }
  typedef std::pair<int64_t, std::function<void()>> JobElem;
  struct Compare {
    bool operator()(const JobElem& lhs, const JobElem& rhs) {
      return std::forward_as_tuple(
                 lhs.first,
                 reinterpret_cast<uintptr_t>(lhs.second.target<std::function<void()>>())) >
             std::forward_as_tuple(
                 rhs.first,
                 reinterpret_cast<uintptr_t>(rhs.second.target<std::function<void()>>()));
    }
  };

  std::priority_queue<JobElem, std::vector<JobElem>, Compare> jobs;
};

class MatchingEngine {
 public:
  MatchingEngine(
      const ::coin2::exchange::order::AccountSimConfig& acct_config)
      : submit_to_first_fill_latency_ns(acct_config.latency_sec() * 1e9),
        cancel_to_confirm_latency_ns((
            acct_config.cancel_latency_sec() > 0 ?
                acct_config.cancel_latency_sec() : acct_config.latency_sec()) * 1e9),
        submit_to_accept_latency_ns(
            acct_config.latency_sec() * 1e9 * 0.5),
        acct_config_(acct_config),
        holder(acct_config) {}
  ~MatchingEngine() {}

  void UpdateTimeOnly(int64_t time) {
    // holder.SetExchangeTime(time);
    // holder.SetFetchedTime(time);
    // NOTE(jhkim): forced to set time. especially when you use timer.
    holder.time = time;
    holder.ftime = time;
    job_queue.UpdateTimeOnly(time);
  }
  void UpdateFeed(const Symbol* symbol, const ::fastfeature::FeedMessage& feed);
  void SendOrder(
      int64_t order_id,
      std::optional<int64_t> last_time,
      double price,
      double qty,
      int sign,
      bool post_only,
      std::string tag,
      bool ioc = false,
      std::optional<double> submit_max_rate_used_ratio = {},
      std::optional<double> cancel_max_rate_used_ratio = {});
  void CancelOrder(int64_t order_id);
  double GetPosition() const { return holder.GetPosition(); }
  double GetWorkingQty(int sign) const { return holder.GetWorkingQty(sign); }
  double GetMtmPrice() const { return holder.GetMtmPrice(); }
  int64_t GetFetchedTime() const { return holder.GetFetchedTime(); }
  const std::vector<OrderRecordProto>& GetOrderRecords() const {
    return holder.GetOrderRecords();
  }
  void ClearOrderRecords() { holder.ClearOrderRecords(); }
  auto GetWorkingBbo() const { return holder.GetWorkingBbo(); }

 private:
  const int64_t submit_to_first_fill_latency_ns = 0;
  const int64_t cancel_to_confirm_latency_ns = 0;
  const int64_t submit_to_accept_latency_ns = 0;

  ::coin2::exchange::order::AccountSimConfig acct_config_;
  SymbolMatchingEngine holder;
  JobQueue job_queue;
};

}  // namespace coin2::exchange::sim::order_v1
